package com.atguigu.bigdata.chapter07.state;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author lzc
 * @Date 2022/9/6 15:03
 */
public class Flink02_Operator_UnionList {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        
        // 每隔3秒中做一次checkpoint
        // 这样理解: 每隔3秒,把程序的所有的状态做一次持久化, 程序出了问题重启的时候, 可以从持久化状态中恢复数据
        env.enableCheckpointing(3000);
        
        // 在flatMap算子中使用算子状态
        SingleOutputStreamOperator<String> stream = env
            .socketTextStream("hadoop162", 9999)
            .flatMap(new MyFlatMapFunction());
        
        stream.addSink(new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                if (value.contains("x")) {
                    throw new RuntimeException("碰到了x, 手动抛出一个异常....");
                }
            }
        });
        
        stream.print();
    
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    public static class MyFlatMapFunction implements FlatMapFunction<String, String>, CheckpointedFunction {
        List<String> words = new ArrayList<>();
        private ListState<String> wordsState;
    
        @Override
        public void flatMap(String line,
                            Collector<String> out) throws Exception {
            // 每行数据用空格切割, 存入到一个List集合中
            for (String word : line.split(" ")) {
                words.add(word);
                
            }
            out.collect(words.toString());
            
        }
    
        // 对状态做快照: 每隔3s中, 把状态中的数据进行一次快照存储(持久化)
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            //每隔3s每个并行度执行一次
//            System.out.println("MyFlatMapFunction.snapshotState");
//            wordsState.clear(); // 清楚地所有的值
//            wordsState.addAll(words);
            
            wordsState.update(words);   // 更新状态(覆盖操作)
            
        }
    
        // 初始化状态: 程序启动的每个并行度执行一次(把快照中的数据加载到状态中)
        @Override
        public void initializeState(FunctionInitializationContext ctx) throws Exception {
            System.out.println("MyFlatMapFunction.initializeState");
            
            // 启动的时候, 把状态中的数据取出, 存入到 ArrayList中
            wordsState = ctx.getOperatorStateStore().getUnionListState(new ListStateDescriptor<String>("wordsState", String.class));
    
            Iterable<String> it = wordsState.get();
            for (String word : it) {
                words.add(word);
            }
        }
    }
}
